package com.pwser.config;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import javax.annotation.PostConstruct;

import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.Binding.DestinationType;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import com.fasterxml.jackson.databind.JavaType;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.pwser.entity.Order;

@Configuration
public class AmqpConfig {
	
	//C客户端消息发布交换器（作为消息延迟消费的缓冲路由）
	@Value("${rabbitmq.exchange.publish}")
	private String EXCHANGE_PUBLISH;
	
	//java客户端消息消费交换器
	@Value("${rabbitmq.exchange.receive}")
	private String EXCHANGE_RECEIVE;
	
	//java客户端消息消费发生异常的交换器
	@Value("${rabbitmq.exchange.exception}")
	private String EXCHANGE_EXCEPTION;
	
	//java客户端消息消费发生异常的公用队列
	@Value("${rabbitmq.exceptionQueue}")
	private String EXCEPTION_QUEUE;
	
	//配电室一对一消息队列的json字符串
	@Value("${rabbitmq.queues}")
	private String RABBITMQ_QUEUES;
	
	//java消费队列的名字数组
	public static String[] QUEUE_RECEIVE_NAMES = null;
	
	@Autowired
	private AmqpAdmin amqpAdmin;
	
	@Autowired
	private ConnectionFactory connectionFactory;
	
	//使用json序列化消息
	@Bean
	public MessageConverter messageConverter() {
		return new Jackson2JsonMessageConverter();
	}
	
	/**
	 * 初始化消息服务
	 * 1、创建三个默认的消息交换器
	 * 2、创建公用的消息异常队列queue.jnpd.exception
	 * 3、循环创建消息队列，会根据配置文件中的队列命名前缀生成与消息交换器对应的两个个队列：
	 * 		queue.jnpd.pd001.publish、queue.jnpd.pd001.receive、
	 * 4、绑定对应的交换器和队列，并设置路由键routeKey
	 */
	@PostConstruct
	public void initRabbitMQ() throws Exception {
		//获取需要初始化的消息队列
		ObjectMapper mapper = new ObjectMapper();
		JavaType javaType = mapper.getTypeFactory().constructParametricType(ArrayList.class, HashMap.class);
		List<Map> queueList = (List<Map>) mapper.readValue(RABBITMQ_QUEUES, javaType);
		
		//创建三个默认的消息交换器
		amqpAdmin.declareExchange(new DirectExchange(EXCHANGE_PUBLISH, true, false));
		amqpAdmin.declareExchange(new DirectExchange(EXCHANGE_RECEIVE, true, false));
		amqpAdmin.declareExchange(new DirectExchange(EXCHANGE_EXCEPTION, true, false));
		
		//创建公用的消息异常队列
		amqpAdmin.declareQueue(new Queue(EXCEPTION_QUEUE, true, false, false, null));
		amqpAdmin.declareBinding(new Binding(EXCEPTION_QUEUE, DestinationType.QUEUE, EXCHANGE_EXCEPTION, "exception", null));
		
		
		//循环创建消息队列
		Map<String, Object> params = new HashMap<>();
		int i = 0;
		QUEUE_RECEIVE_NAMES = new String[queueList.size()];
		for(Map queueMap : queueList) {
	        //x-dead-letter-exchange 声明了队列里的死信转发到的DLX名称，
	        params.put("x-dead-letter-exchange", EXCHANGE_RECEIVE);
	        //x-dead-letter-routing-key 声明了这些死信在转发时携带的 routing-key 名称。
	        params.put("x-dead-letter-routing-key", queueMap.get("routeKey"));
	        //创建消息队列，并指明死信的交换器
			amqpAdmin.declareQueue(new Queue(queueMap.get("prefixName")+".publish", true, false, false, params));
			//绑定对列与交换器
			amqpAdmin.declareBinding(new Binding(queueMap.get("prefixName")+".publish", DestinationType.QUEUE, EXCHANGE_PUBLISH, queueMap.get("routeKey").toString(), null));
			
			params.put("x-dead-letter-exchange", EXCHANGE_EXCEPTION);
			params.put("x-dead-letter-routing-key", "exception");
			amqpAdmin.declareQueue(new Queue(queueMap.get("prefixName")+".receive", true, false, false, params));
			amqpAdmin.declareBinding(new Binding(queueMap.get("prefixName")+".receive", DestinationType.QUEUE, EXCHANGE_RECEIVE, queueMap.get("routeKey").toString(), null));
			QUEUE_RECEIVE_NAMES[i++] = queueMap.get("prefixName")+".receive";
		}
		
	}
	
	@Bean
    public SimpleMessageListenerContainer receiveMessageListenerContainer() {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setDefaultRequeueRejected(false); //异常会丢弃消息或者重新发布到死信队列中
        //设置监听的队列名
        container.setQueueNames(QUEUE_RECEIVE_NAMES);
        container.setMessageListener(receiveQueuesListener());
        return container;
    }
	
	@Bean
	public MessageListener receiveQueuesListener() {
		return new MessageListener() {
			@Override
			public void onMessage(Message message) {
				int i = 1 / 0;
				System.out.println("###########################################");
				System.out.println("【receiveQueue 监听的消息】 - 【消费时间】 - ["+new Date()+"]- 【消息内容】 - ["+message+"]");
				Order order = null;
				try {
					ObjectMapper mapper = new ObjectMapper();
					order = (Order) mapper.readValue(message.getBody(), Order.class);
				} catch (Exception e) {
					e.printStackTrace();
				}
				//业务逻辑
				if (order.getOrderStatus() == 0) {
					order.setOrderStatus(2);
					System.out.println("【该订单未支付，取消订单】" + order.toString());
				} else if (order.getOrderStatus() == 1) {
					System.out.println("【该订单已完成支付】");
				} else if (order.getOrderStatus() == 2) {
					System.out.println("【该订单已取消】");
				}
				System.out.println("###########################################");
			}
		};
	}
	
}
